package org.example.flux;

import org.springframework.http.MediaType;
import org.springframework.stereotype.Controller;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.ResponseBody;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Sinks;
import reactor.core.scheduler.Schedulers;

/**
 * flux流
 * <p>Spring-FluxWeb支持的响应式流处理</p>
 */
@Controller
@RequestMapping("")
public class Demo1Control {
    //插槽
    private final Sinks.Many<Integer> sink = Sinks.many().multicast().onBackpressureBuffer();

    /**
     * 页面
     * <p>localhost:8000</p>
     *
     * @return 页面
     */
    @GetMapping("")
    public String index() {
        return "flux";
    }

    /**
     * 获取flux流
     *
     * @return flux流
     */
    @GetMapping(value = "getFlux1", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    @ResponseBody
    public Flux<String> getFlux1() {
        //创建0-1000集合,并遍历
        return Flux.range(0, 1000).publishOn(Schedulers.boundedElastic()).handle((i, sink) -> {
            //休眠70秒,模拟被动响应/阻塞响应
            try {
                Thread.sleep(100);
            } catch (InterruptedException ignored) {
            }
            //执行,生成随机汉字
            sink.next(String.valueOf((char) (0x4e00 + (int) (Math.random() * (0x9fa5 - 0x4e00 + 1)))));
        });
    }


    /**
     * 获取flux流带插槽
     *
     * @return flux流带插槽
     */
    @GetMapping(value = "getFlux2", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    @ResponseBody
    public Flux<Integer> getFlux2() {
        //新线程异步执行
        new Thread(() -> {
            //发布
            for (int i = 10; i >= 0; i--) {
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
                sink.emitNext(i, Sinks.EmitFailureHandler.FAIL_FAST);
            }
        }).start();
        return sink.asFlux();
    }
}
